package com.atguigu.edu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.beans.VideoPlaySecBean;
import com.atguigu.edu.realtime.utils.DateFormatUtil;
import com.atguigu.edu.realtime.utils.MyClickhouseUtil;
import com.atguigu.edu.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;

/**
 * @Classname DwsPlayVideoPlayTimes
 * @Date 2022/11/20 14:23
 * @Created by arun
 * 当日各章节播放时长
 */
public class DwsPlayVideoPlaySec {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        String topic = "dwd_play_videoPlayTimes";
        String groupId = "dws_play_videoPlaySec";
        FlinkKafkaConsumer<String> kafkaConsumer = MyKafkaUtil.getKafkaConsumer(topic, groupId);
        DataStreamSource<String> kafkaStrDS = env.addSource(kafkaConsumer);
        kafkaStrDS.print("kafkaStrDS");
        // {"video_id":4184,"uid":"2772","mid":"mid_171","sid":"257978f1-5554-4210-9094-ba213421320d",
        // "play_sec":"30","position_sec":"360","ts":"1668732144837","chapterId":25079}

        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.map(JSON::parseObject);
        jsonObjDS.print("jsonObjDS");
        // {"uid":"2772","play_sec":"30","chapterId":25079,"position_sec":"360","mid":"mid_171","video_id":4184,
        // "sid":"257978f1-5554-4210-9094-ba213421320d","ts":"1668732144837"}

        SingleOutputStreamOperator<VideoPlaySecBean> videoPlaySecBean = jsonObjDS.map(
                new MapFunction<JSONObject, VideoPlaySecBean>() {
                    @Override
                    public VideoPlaySecBean map(JSONObject jsonObj) throws Exception {
                        VideoPlaySecBean videoPlaySecBean = VideoPlaySecBean.builder()
                                .chapterId(jsonObj.getString("chapterId"))
                                .playSec(jsonObj.getLong("play_sec"))
                                .ts(jsonObj.getLong("ts"))
                                .build();
                        return videoPlaySecBean;
                    }
                }
        );
        // videoPlaySecBean.print("videoPlaySecBean");

        SingleOutputStreamOperator<VideoPlaySecBean> withWatermarkDS = videoPlaySecBean.assignTimestampsAndWatermarks(
                WatermarkStrategy.<VideoPlaySecBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(new SerializableTimestampAssigner<VideoPlaySecBean>() {
                            @Override
                            public long extractTimestamp(VideoPlaySecBean element, long recordTimestamp) {
                                return element.getTs();
                            }
                        })
        );

        // 按章节ID进行分组
        KeyedStream<VideoPlaySecBean, String> keyedDS = withWatermarkDS.keyBy(VideoPlaySecBean::getChapterId);
        // keyedDS.print("keyedDS");

        // 开窗
        WindowedStream<VideoPlaySecBean, String, TimeWindow> windowDS = keyedDS.window(TumblingEventTimeWindows.of(Time.seconds(10)));

        // 聚合
        SingleOutputStreamOperator<VideoPlaySecBean> reduceDS = windowDS.reduce(
                new ReduceFunction<VideoPlaySecBean>() {
                    @Override
                    public VideoPlaySecBean reduce(VideoPlaySecBean value1, VideoPlaySecBean value2) throws Exception {
                        value1.setPlaySec(value1.getPlaySec() + value2.getPlaySec());
                        return value1;
                    }
                },
                new WindowFunction<VideoPlaySecBean, VideoPlaySecBean, String, TimeWindow>() {
                    @Override
                    public void apply(String groupId, TimeWindow window, Iterable<VideoPlaySecBean> input, Collector<VideoPlaySecBean> out) throws Exception {
                        String stt = DateFormatUtil.toYmdHms(window.getStart());
                        String end = DateFormatUtil.toYmdHms(window.getEnd());
                        for (VideoPlaySecBean secBean : input) {
                            secBean.setStt(stt);
                            secBean.setEdt(end);
                            out.collect(secBean);
                        }
                    }
                }
        );
        reduceDS.print(">>>");

        // 将聚合结果写到ck
        reduceDS.addSink(
                MyClickhouseUtil.getSinkFunction("insert into dws_play_chapter_videoPlaySec_window values(?,?,?,?,?)")
        );

        env.execute();
    }
}
